package com.surfilter.massdata.spark.util.ipcheck;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.Set;

import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.SQLContext;

import com.google.common.collect.Lists;
import com.google.common.collect.Sets;
import com.surfilter.massdata.spark.bean.ipcheck.ErrIp;
import com.surfilter.massdata.spark.bean.ipcheck.IpBean;
import com.surfilter.massdata.spark.bean.ipcheck.ObjectInfo;

import scala.Tuple2;
import scala.Tuple3;

/**
 * @author hutao
 * @date 创建时间：2017年6月8日 下午5:14:10
 * @Description：上报库VS活跃IP
 * @version 1.0
 * 
 **/
public class IpCheckUploadVsActiveIp {
	public static void dealBusiness(JavaRDD<IpBean> ipBeanRdd, SQLContext sc) throws Exception {
		// JavaRDD<IpBean>转换为JavaPairRDD<Long, Iterable<IpBean>>为了对ip进行分组
		JavaPairRDD<Long, Iterable<IpBean>> pariRDD = IpCheckUtil.transJavaPariRdd(ipBeanRdd);
		JavaSparkContext jsc = JavaSparkContext.fromSparkContext(sc.sparkContext());
		final Broadcast<Map<Long, String>> broadBbdwMap = jsc.broadcast(IpCheckUtil.getBr3010IdMap());//广播bbdwId列表
		 JavaRDD<ErrIp> conflictRdd=pariRDD.flatMapToPair(new PairFlatMapFunction<Tuple2<Long,Iterable<IpBean>>, String, Long>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<Tuple2<String, Long>> call(Tuple2<Long, Iterable<IpBean>> tuple) throws Exception {
				long ip = tuple._1;
				Iterable<IpBean> its = tuple._2;
				List<IpBean> sourceList = Lists.newArrayList();
				List<IpBean> uploadList = Lists.newArrayList();
				Set<ErrIp> errIpSet=Sets.newHashSet();
				Set<Tuple2<String,Long>> resultTupleSet=Sets.newHashSet();
				IpBean isIpBean = null;//存活跃数据
				IpBean tailIpBean = null;//存尾巴
				Map<Long, String> bbdwMap=broadBbdwMap.getValue();
				for (IpBean iterBean : its) {
					IpBean ipBean = new IpBean();
					ipBean = ConvertUtil.copy(ipBean, iterBean);
					if(ipBean.getStorageType().equals(IpCheckConstants.ACTIVEIP)){
						isIpBean = ipBean;
					}else{
						if(ipBean.getTflag().equals(IpCheckConstants.IP_SOURCE_TYPE)){
							sourceList.add(ipBean);
						}else{
							//尾巴
							if(!ipBean.getFpfs().equals(IpAllocateType.RE_ALLOCATE.getTypeValue())){
								tailIpBean = ipBean;
							}
							//再分配
							uploadList.add(ipBean);
						}
					}
				}
				
				if(isIpBean != null){
					long bbdw = isIpBean.getBbdw_id();
					
					if(uploadList.size() != 0){
						if(tailIpBean != null){
							//有尾巴
							//先比机房或者企业是否相同，再比分配方式是否是预留 在江苏放过对机房的比对
							if(bbdw == tailIpBean.getBbdw_id()){
								if(tailIpBean.getFpfs() == null || tailIpBean.getFpfs().equals(IpAllocateType.RESERVE.getTypeValue())){
									ErrIp errIp = new ErrIp();
									errIp.setBbdws(tailIpBean.getBbdw_id().toString());
									errIp.setRegionIds(IpCheckConstants.ACTIVEIP + "_" + isIpBean.getId() + "," + tailIpBean.getStorageType() + "_" + tailIpBean.getTflag() + "_" + tailIpBean.getId());
									errIp.setConflictType(ConflictType.IP_UPLOAD_USE_ERROR.getTypeValue());
									errIp.setStorageType(IpCheckConstants.ACTIVEIP);
									errIp.setJsonStr(tailIpBean.getConflictMsg());
									errIpSet.add(errIp);
								}
							}else{
								if(tailIpBean.getBbdw_id().equals(isIpBean.getBbdw_id())){    //????
									ErrIp errIp = new ErrIp();
									errIp.setBbdws(tailIpBean.getBbdw_id().toString());
									errIp.setRegionIds(IpCheckConstants.ACTIVEIP + "_" + isIpBean.getId() + "," + tailIpBean.getStorageType() + "_" + tailIpBean.getTflag() + "_" + tailIpBean.getId());
									errIp.setConflictType(ConflictType.IP_ACTIVEIP_UPLOAD_CONFLICT_ERROR.getTypeValue());
									errIp.setStorageType(IpCheckConstants.ACTIVEIP);
									errIp.setJsonStr(isIpBean.getConflictMsg()+"\001"+tailIpBean.getConflictMsg());
									errIpSet.add(errIp);
								}else{
									ErrIp errIp = new ErrIp();
									errIp.setBbdws(isIpBean.getBbdw_id().toString() + "," + tailIpBean.getBbdw_id().toString());
									errIp.setRegionIds(IpCheckConstants.ACTIVEIP + "_" + isIpBean.getId() + "," + tailIpBean.getStorageType() + "_" + tailIpBean.getTflag() + "_" + tailIpBean.getId());
									errIp.setConflictType(ConflictType.IP_ACTIVEIP_UPLOAD_CONFLICT_ERROR.getTypeValue());
									errIp.setStorageType(IpCheckConstants.ACTIVEIP);
									errIp.setJsonStr(isIpBean.getConflictMsg()+"\001"+tailIpBean.getConflictMsg());
									errIpSet.add(errIp);
								}
							}
						}else{
							//没有尾巴 那就是分配到外省了 作为冲突下发
							for(IpBean outer : uploadList){
								if(outer.getFpdx() != null && bbdwMap.get(outer.getFpdx()) == null){
									 if(outer.getBbdw_id().equals(isIpBean.getBbdw_id())){
										 ErrIp errIp = new ErrIp();
										 errIp.setBbdws(outer.getBbdw_id().toString());
										 errIp.setRegionIds(IpCheckConstants.ACTIVEIP + "_" + isIpBean.getId() + "," + outer.getStorageType() + "_" + outer.getTflag() + "_" + outer.getId());
										 errIp.setConflictType(ConflictType.IP_ACTIVEIP_UPLOAD_CONFLICT_ERROR.getTypeValue());
										 errIp.setStorageType(IpCheckConstants.ACTIVEIP);
										 errIp.setJsonStr(isIpBean.getConflictMsg()+"\001"+outer.getConflictMsg());
										 errIpSet.add(errIp);
									 }else{
										 ErrIp errIp = new ErrIp();
										 errIp.setBbdws(isIpBean.getBbdw_id().toString() + "," + outer.getBbdw_id().toString());
										 errIp.setRegionIds(IpCheckConstants.ACTIVEIP + "_" + isIpBean.getId() + "," + outer.getStorageType() + "_" + outer.getTflag() + "_" + outer.getId());
										 errIp.setConflictType(ConflictType.IP_ACTIVEIP_UPLOAD_CONFLICT_ERROR.getTypeValue());
										 errIp.setStorageType(IpCheckConstants.ACTIVEIP);
										 errIp.setJsonStr(isIpBean.getConflictMsg()+"\001"+outer.getConflictMsg());
										 errIpSet.add(errIp);
									 }
								}
							}
							
						}
					}else{
						//上报库没有对应ip 在江苏放过这种冲突
//						ErrIp errIp = new ErrIp();
//						errIp.setBbdws(isIpBean.getBbdw_id().toString());
//						errIp.setRegionIds(IpCheckConstants.ACTIVEIP + "_" + isIpBean.getId());
//						errIp.setConflictType(ConflictType.IP_ACTIVE_OMIT_ERROR.getTypeValue());
//						errIp.setStorageType(IpCheckConstants.ACTIVEIP);
//						this.mergeConflictIpSeg(errIp, ip, mos, log, context);
					}
				}
				for(ErrIp errIp:errIpSet){
					resultTupleSet.add(new Tuple2<String,Long>(errIp.toString(),ip));
				}
				return resultTupleSet;
			}
		}).groupByKey().flatMap(new FlatMapFunction<Tuple2<String,Iterable<Long>>, ErrIp>() {

			private static final long serialVersionUID = 1L;

			@Override
			public Iterable<ErrIp> call(Tuple2<String, Iterable<Long>> tuple2) throws Exception {
				String conflictStr=tuple2._1;
				Iterable<Long> its=tuple2._2;
				List<ErrIp> resultList=new ArrayList<ErrIp>();
				String[] array=conflictStr.split("\t");
				ErrIp errIp = new ErrIp();
				if(array.length==5){
					errIp = ConvertUtil.convertErrIp(array, errIp);
				}
				
				List<Long> list=new ArrayList<Long>();
				for(Long ip:its){
					list.add(ip);
				}
				List<String> segList=IpUtils.getIpSeg(list);
				for(String seg:segList){
					long qsip=Long.parseLong(seg.split(",")[0]);
					long zzip=Long.parseLong(seg.split(",")[1]);
					ErrIp source = ConvertUtil.copyErrIp(new ErrIp(), errIp);
					source.setQsip(qsip);
					source.setZzip(zzip);
					resultList.add(source);
				}
				return resultList;
			}
		});
		 DataFrame conlinctDF=sc.createDataFrame(conflictRdd, ErrIp.class);
		 IpCheckUtil.writeConflictToTable(conlinctDF);//将冲突数据入库
	}

}
